package cn.iocoder.yudao.module.forward.mqtt.broker;

import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.module.forward.controller.admin.eew.vo.EewPageReqVO;
import cn.iocoder.yudao.module.forward.controller.admin.eqr.vo.EqrPageReqVO;
import cn.iocoder.yudao.module.forward.dal.dataobject.eew.EewDO;
import cn.iocoder.yudao.module.forward.dal.dataobject.eqr.EqrDO;
import cn.iocoder.yudao.module.forward.mqtt.*;
import cn.iocoder.yudao.module.forward.mqtt.message.AckMessage;
import cn.iocoder.yudao.module.forward.service.eew.EewService;
import cn.iocoder.yudao.module.forward.service.eqr.EqrService;
import cn.iocoder.yudao.module.forward.service.scope.ScopeService;
import cn.iocoder.yudao.module.forward.service.strategy.StrategyService;
import com.alibaba.fastjson.JSONObject;
import lombok.Data;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Map;

@Slf4j
@Service
public class BrokerMessageHandler implements MessageHandler {
    @Resource(name = "SourceMqttManager")
    private MqttManager mqttManager;
    @Resource
    private ScopeService mScopeService;
    @Resource
    private StrategyService mStrategyService;
    @Resource
    private EewService mEewService;
    @Resource
    private EqrService mEqrService;

    @Override
    public void handleMessage(MqttInfo mqttInfo, String topic, String data) throws Exception {
        String topicType = (String) mqttInfo.getTopics().get(topic);
        if (topicType == null) {
            log.error("{}({})不存在该主题:{}", mqttManager.getTag(), mqttInfo.getName(), topic);
            return;
        }

        switch (topicType) {
            case MqttConstants.TOPIC_TYPE_ACK:
                log.info("收到回复消息:{}", data);
                onAck(mqttInfo, topic, topicType, data);
                break;
        }
    }

    /**
     * 回复消息，回复消息应该只存放
     */
    private void onAck(MqttInfo mqttInfo, String topic, String topicType, String data) {
        AckMessage ackMessage = JSONObject.parseObject(data, AckMessage.class);
        AckResult ackResult = new AckResult();
        ackResult.setBrokerId(mqttInfo.getId());
        ackResult.setOrgTopic(topic);
        ackResult.setOrgMsg(data);
        ackResult.setMsg(ackMessage);
        ackResult.setTime(LocalDateTime.now());

        switch (ackMessage.getMsgType()) {
            case MqttConstants.MSG_NORMAL:
                onEewNormal(ackResult, topicType, mqttInfo);
                break;
            case MqttConstants.MSG_NORMAL_EXERCISE:
                break;
            case MqttConstants.MSG_NORMAL_TEST:
                break;
            case MqttConstants.MSG_FAST_TEST:
                break;
            case MqttConstants.MSG_FAST_AUTO:
            case MqttConstants.MSG_FAST_FORMAL:
                onEqr(ackResult, topicType, mqttInfo);
                break;
        }
    }

    /**
     * 普通预警消息回复
     *
     * @param ackResult
     * @param topicType
     * @param mqttInfo
     */
    private void onEewNormal(AckResult ackResult, String topicType, MqttInfo mqttInfo) {
        EewPageReqVO eewPageReqVO = new EewPageReqVO();
        eewPageReqVO.setBrokerId(ackResult.getBrokerId());
        eewPageReqVO.setMsgId(ackResult.getMsg().getMsgID());
        try {
            PageResult<EewDO> eewPage = mEewService.getEewPage(eewPageReqVO);
            if (eewPage.getTotal() != 1) {
                // TODO: 2025/5/8 记录错误   同一条消息在一个转发下面应该只有一条数据
                throw new RuntimeException("数据存储错误, 转发(" + mqttInfo.getName() + "), msgID(" + ackResult.getMsg().getMsgID() + ")");
            }

            EewDO eewDO = eewPage.getList().get(0);
            MqttConnect mqttConnect = mqttManager.getMqttById(eewDO.getSourceId());
            if (mqttConnect == null) {
                // TODO: 2025/5/8 记录错误
                throw new RuntimeException("信息源(" + mqttConnect.getMqttInfo().getName() + ")不存在");
            }
            ackResult.setSourceId(eewDO.getSourceId());
            Map<String, Object> topics = mqttConnect.getMqttInfo().getTopics();
            String ackTopic = (String) topics.get(MqttConstants.TOPIC_TYPE_ACK);
            if (ackTopic != null) {
                String data = beforePublish(ackResult.getOrgMsg());
                mqttConnect.publish(ackTopic, data, MqttConstants.MQTT_QOS);
                ackResult.setForwardTopic(ackTopic);
                ackResult.setForwardMsg(data);
            }

            // TODO: 2025/5/8 记录回复消息 ackResult
            //如果是第一次收到该msgId的消息  则开启5分钟定时任务   5分钟后进行第二次回复
            //String ackTopic2 = (String) topics.get(MqttConstants.TOPIC_TYPE_ACK2);
        } catch (Exception e) {
            // TODO: 2025/5/8 记录错误信息
        }
    }

    /**
     * 速报消息回复   自动跟正式未区分
     *
     * @param ackResult
     */
    private void onEqr(AckResult ackResult, String topicType, MqttInfo mqttInfo) {
        EqrPageReqVO eqrPageReqVO = new EqrPageReqVO();
        eqrPageReqVO.setBrokerId(ackResult.getBrokerId());
        eqrPageReqVO.setMsgId(ackResult.getMsg().getMsgID());
        try {
            PageResult<EqrDO> eqrPage = mEqrService.getEqrPage(eqrPageReqVO);
            if (eqrPage.getTotal() != 1) {
                // TODO: 2025/5/8 记录错误   同一条消息在一个转发下面应该只有一条数据
                throw new RuntimeException("数据存储错误, 转发(" + mqttInfo.getName() + "), msgID(" + ackResult.getMsg().getMsgID() + ")");
            }

            EqrDO eqrDO = eqrPage.getList().get(0);
            MqttConnect mqttConnect = mqttManager.getMqttById(eqrDO.getSourceId());
            if (mqttConnect == null) {
                // TODO: 2025/5/8 记录错误
                throw new RuntimeException("信息源(" + mqttConnect.getMqttInfo().getName() + ")不存在");
            }
            ackResult.setSourceId(eqrDO.getSourceId());
            Map<String, Object> topics = mqttConnect.getMqttInfo().getTopics();
            String ackTopic = (String) topics.get(MqttConstants.TOPIC_TYPE_ACK);
            if (ackTopic != null) {
                String data = beforePublish(ackResult.getOrgMsg());
                ackResult.setForwardTopic(ackTopic);
                ackResult.setForwardMsg(data);
            }

            // TODO: 2025/5/8 记录回复消息
            //如果是第一次收到该msgId的消息  则开启5分钟定时任务   5分钟后进行第二次回复
            //String ackTopic2 = (String) topics.get(MqttConstants.TOPIC_TYPE_ACK2);
        }catch (Exception e){
            // TODO: 2025/5/8 记录错误信息
        }
    }

    /**
     * 发送前处理
     *
     * @param data
     * @return
     */
    private String beforePublish(String data) {
        return data;
    }

    @Data
    @ToString
    class AckResult {
        private String orgTopic;
        private String orgMsg;
        private String forwardTopic;
        private String forwardMsg;
        private AckMessage msg;
        private LocalDateTime time;
        private Integer sourceId;
        private Integer brokerId;
    }
}
